package com.relytosoft.mizar.analyze.strategy.consumer;

import com.alibaba.fastjson.JSON;
import com.relytosoft.dubhe.common.core.constant.CommonConstants;
import com.relytosoft.dubhe.common.core.object.ResponseResult;
import com.relytosoft.dubhe.common.kafka.producer.KafkaOpMsg;
import com.relytosoft.dubhe.common.kafka.producer.KafkaProducer;
import com.relytosoft.mizar.analyze.base.constant.MediaMqConstants;
import com.relytosoft.mizar.analyze.base.entity.pojo.dto.DispositionReqDto;
import com.relytosoft.mizar.analyze.strategy.algorithm.face.service.FaceService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * DispositionConsumer
 *
 * @Author SunLi
 * @Date: 2022/5/17 14:19
 * @Version 1.0
 */
@Component
public class DispositionConsumer {
    protected Logger log = LoggerFactory.getLogger(this.getClass());

    @Resource
    private FaceService faceService;

    @Resource
    private KafkaProducer kafkaProducer;

    @KafkaListener(id = "consumer-analyze-face-disposition", topics = MediaMqConstants.FACE_DISPOSITION_TOPIC, containerFactory = "singleContainerFactory")
    public void processSingleMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        try {
            boolean isConsume = false;
            KafkaOpMsg kafkaOpMsg = JSON.parseObject(record.value().toString(), KafkaOpMsg.class);
            String data = kafkaOpMsg.getData();
            switch (kafkaOpMsg.getOpType()) {
                case "insert":
                    isConsume = createDisposition(data);
                    break;
                case "delete":
                    isConsume = deleteDisposition(data);
                default:
                    break;
            }
            log.info("线程:{} 消费-人脸布控同步->{}", Thread.currentThread(), data);
            if (isConsume) {
                ack.acknowledge();
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    /**
     * 人脸布控任务添加
     *
     * @param data 布控信息
     * @return 消费结果
     */
    private boolean createDisposition(String data) {
        DispositionReqDto dispositionReqDto = JSON.parseObject(data, DispositionReqDto.class);
        ResponseResult result = faceService.createDispostion(dispositionReqDto);
        if (result.getCode() == CommonConstants.SUCCESS) {
            log.info("线程:{} 消费-人脸布控同步结果->{}", Thread.currentThread(), result.getData());
            KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
            kafkaOpMsg.setData(JSON.toJSONString(result.getData()));
            kafkaProducer.send(MediaMqConstants.FACE_ANALYZE_DISPOSITION_ID_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
            return true;
        }
        return false;
    }

    /**
     * 人脸布控任务删除
     *
     * @param data 布控信息
     * @return 消费结果
     */
    private boolean deleteDisposition(String data) {
        ResponseResult result = faceService.deleteDispostion(data);
        if (result.getCode() == CommonConstants.SUCCESS) {
            return true;
        }
        return false;
    }
}
